package com.stefanie.kafkademo.demo;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.*;
import java.util.regex.Pattern;

/**
 * @author: stefanie-zy
 * @time: 2023/12/11-15:48
 * @description: kafka-消费者
 */
@Slf4j
public class MyConsumer {

    private final static String TOPIC_NAME = "stefanie-zy-test"; // 主题
    private final static String CONSUMER_GROUP_NAME = "testGroup"; // 消费者组
    private final static Boolean ENABLE_AUTO_COMMIT_CONFIG_STR = Boolean.TRUE; // 消费者提交模式

    public static void main(String[] args) {
        // 存放键值对属性
        Properties properties = new Properties();

        // 设置生产者ID
//        properties.put(ProducerConfig.CLIENT_ID_CONFIG, "");

        // 集群使用逗号隔开
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");

        // 分组配置
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_NAME);

        // 设置Key、Value的序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // 新的消费组消费消息
        // 1、"earliest":第一次从头开始消费消息，以后按照offset进行消费。区别于consumer.seekToBeginning()。
        // 2、"latest":只消费启动之后的数据。
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // 心跳配置，心跳上报时间
        properties.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000);
        // 心跳超过10s没有上报，kafka会将消费者踢出消费组，把分区分配给其他消费者
        properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10 * 1000);

        // 性能调优
        // poll一次性拉取消息个数
        properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
        // 两次poll消息的时间如果超过30s的间隔[指的是业务处理时间较慢]，kafka会认为其消费能力较弱，踢出消费组，将分区分配给其他消费者
        properties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30 * 1000);

        // 设置自动提交
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, ENABLE_AUTO_COMMIT_CONFIG_STR); // 设置自动提交
        if (!ENABLE_AUTO_COMMIT_CONFIG_STR) {
            properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000"); // 设置自动提交间隔时间，单位：ms}
        }

        // 创建消费者客户端
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);

        // 订阅主题列表
        consumer.subscribe(List.of(TOPIC_NAME));
        // 订阅多个主题也可以使用正则表达式的方式[订阅所有以topic开头的主题]
        consumer.subscribe(Pattern.compile("topic-.*"));

        // 指定消费
        // 指定分区消费[0分区]
//        consumer.assign(List.of(new TopicPartition(TOPIC_NAME, 0)));
//        // 回溯消费[从分区的offset=0的位置开始消费]
//        consumer.assign(List.of(new TopicPartition(TOPIC_NAME, 0)));
//        consumer.seekToBeginning(List.of(new TopicPartition(TOPIC_NAME, 0)));
//        // 指定offset消费[offset=10]
//        consumer.assign(List.of(new TopicPartition(TOPIC_NAME, 0)));
//        consumer.seek(new TopicPartition(TOPIC_NAME, 0), 10);
//        // 指定时间节点消费
//        consumerByTime(consumer, new Date());

        while (true) {
            // 长轮询拉取数据[设置拉取时间]
            ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(1000)); // 毫秒

            // 逻辑处理
            consumerRecords.forEach(consumerRecord -> log.info("kafka消费者收到消息：partition:{}；offset:{}；key:{}；value:{}。", consumerRecord.partition(), consumerRecord.offset(), consumerRecord.key(), consumerRecord.value()));
            // 提交
            if (ENABLE_AUTO_COMMIT_CONFIG_STR) {
                str(consumer, consumerRecords);
                continue;
            }
            asm(consumer, consumerRecords);
        }
    }


    /**
     * 同步提交
     */
    private static void str(KafkaConsumer<String, String> consumer, ConsumerRecords<String, String> consumerRecords) {
        if (consumerRecords.count() > 0) {
            // 手动-同步提交，当前线程会一直阻塞到offset提交完成之后
            consumer.commitAsync();
        }
    }

    /**
     * 异步提交
     */
    private static void asm(KafkaConsumer<String, String> consumer, ConsumerRecords<String, String> consumerRecords) {
        if (consumerRecords.count() > 0) {
            consumer.commitAsync((offsetAndMetadataMap, e) -> {
                if (e != null) {
                    log.error("手动异步提交失败，offset位：{}", offsetAndMetadataMap);
                    log.error("异常信息：{}", (Object) e.getStackTrace());
                    e.printStackTrace();
                }
            });
        }
    }

    /**
     * 获取目标时间内的消息
     */
    private static void consumerByTime(KafkaConsumer<String, String> consumer, Date targetDate) {
        // 获取主题所有分区信息
        List<PartitionInfo> partitionInfos = consumer.partitionsFor(TOPIC_NAME);
        // 目标时间戳
        long time = targetDate.getTime();
        Map<TopicPartition, Long> map = new HashMap<>();
        // 根据时间确定偏移量offset
        partitionInfos.forEach(par -> map.put(new TopicPartition(TOPIC_NAME, par.partition()), time));
        Map<TopicPartition, OffsetAndTimestamp> parMap = consumer.offsetsForTimes(map);
        parMap.forEach((key, value) -> {
            if (key != null && value != null) {
                long offset = value.offset();
                log.info("partition-{}|offset-{}", key.partition(), offset);
                consumer.assign(List.of(key));
                consumer.seek(key, offset);
            }
        });

    }
}
